package com.bobo.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.bobo.gmall.realtime.app.function.CustomerDeserialization;
import com.bobo.gmall.realtime.app.function.MyJdbcSink;
import com.bobo.gmall.realtime.utils.MyKafkaUtil;
import com.bobo.gmall.realtime.app.function.TableProcessFunction;
import com.bobo.gmall.realtime.bean.TableProcess;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;



//数据流：web/app -> nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka(dwd)/Phoenix(dim)
//程  序：           mockDb -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDBApp -> Kafka/Phoenix(hbase,zk,hdfs)
public class BaseDBApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        env.setStateBackend(new FsStateBackend("hdfs://192.168.45.132:9000/gmall-flink/ck"));
//        env.enableCheckpointing(5000L);   // 即5s触发一次checkPoint
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);  // 模式
//        env.getCheckpointConfig().setAlignmentTimeout(10000L);   // 超时时间 10s
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);    //
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);   // 两次checkpoint间最小间隔时间
//        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart());  老版本需要注意

        //TODO 2.消费kafka（ods_base_db）主题数据创建流
        String sourceTopic = "ods_base_db";
        String groupId = "base_db_app";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将每行数据转换为JSON对象并过滤（delete）  --主流
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //取出数据的操作类型
                        String type = value.getString("type");
                        return !"delete".equals(type);
                    }
                });

        //TODO 4.使用FlinkCDC消费配置表并处理成  --广播流   （即把配置信息表 变为 实时表，随时广播）
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("192.168.45.132")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall_realtime")
                .tableList("gmall_realtime.table_process")
                .deserializer(new CustomerDeserialization())
//                .debeziumProperties()
                .startupOptions(StartupOptions.initial())   //initial(): 无checkPoint时，都是从头开始读;    latest(): 直接读最新的数据
                .build();
        DataStreamSource<String> tableProcessStrDS = env.addSource(sourceFunction);
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor<String, TableProcess>
                ("map-state", String.class, TableProcess.class);
        BroadcastStream<String> broadcastStream = tableProcessStrDS.broadcast(mapStateDescriptor);


        //TODO 5.连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjDS.connect(broadcastStream);


        //TODO 6.分流，处理数据  广播流数据，主流数据
        /**
         * 广播流：（如果有变化，即马上建表，并广播出去让实时计算感知到）
         *   1 解析数据  String -> TableProcess
         *   2 检查Hbase表是否存在并建表
         *   3 写入状态
         * 主流：
         *   1 读取状态（广播）
         *   2 过滤数据
         *   3 分流
         */
        OutputTag<JSONObject> mysqlTag = new OutputTag<JSONObject>("mysql-tag"){};
        SingleOutputStreamOperator<JSONObject> kafka = connectedStream.process(new TableProcessFunction(
                mysqlTag,mapStateDescriptor
        ));

        //TODO 7.提取Kafka流数据 和 Hbase流数据
        DataStream<JSONObject> mysql = kafka.getSideOutput(mysqlTag);

        //TODO 8.将kafka数据写入kafka主题，将Hbase数据写入Phoenix表
        kafka.print("Kafka>>>>>>>>>>>>>>");
        mysql.print("Mysql>>>>>>>>>>>>>>");

        mysql.addSink(new MyJdbcSink());
        kafka.addSink(MyKafkaUtil.getKafkaProducer(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(element.getString("sinkTable"),
                        element.getString("after").getBytes());
            }
        }));

        //TODO 9.启动任务
        env.execute("BaseDBApp");

    }
}
